package org.lwd.datalink.zookeeper;

import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.apache.commons.lang.StringUtils;
import org.lwd.datalink.errors.DatalinkException;

/**
 * @Author: weidong.lin
 */
public class DLinkZkUtils {
    private ZkClientX zkClient;
    private String zkRoot;

    private DLinkZkUtils() {
    }

    public String zkRoot() {
        if (StringUtils.isEmpty(zkRoot)) {
            throw new IllegalStateException("get zkRoot failed,should initial DLinkZkUtils first.");
        }
        return zkRoot;
    }

    public ZkClientX zkClient() {
        if (zkClient == null) {
            throw new IllegalStateException("get zkClient failed,should initial DLinkZkUtils first.");
        }
        return zkClient;
    }

    //--------------------------------------- Static Fields & Methods ---------------------------------------------

    private static DLinkZkUtils zkUtils;

    public static synchronized DLinkZkUtils init(ZkConfig zkConfig, String zkRoot) {
        if (zkUtils != null) {
            throw new DatalinkException("ZkUtils is already initialized,can't initial again.");
        }

        zkUtils = new DLinkZkUtils();
        zkUtils.zkRoot = zkRoot;
        zkUtils.zkClient = ZkClientX.getZkClient(zkConfig);
        return zkUtils;
    }

    public static DLinkZkUtils get() {
        if (zkUtils == null) {
            throw new IllegalStateException("get ZkUtils failed,should initial first.");
        }
        return zkUtils;
    }

    /**
     * 更新指定task的位点信息
     *
     * @param taskId
     * @param data
     * @param path
     */
    public static void updateTask(String taskId, byte[] data, String path) {
        DLinkZkUtils zkUtils = DLinkZkUtils.get();
        try {
            zkUtils.zkClient().writeData(path, data);
        } catch (ZkNoNodeException e) {
            zkUtils.zkClient().createPersistent(path, data, true);// 第一次节点不存在，则尝试创建
        }
    }

    /**
     * 获取指定Task的位点信息，如果没有则返回null
     *
     * @param taskId
     * @param path
     * @return
     */
    public static byte[] getTask(String taskId, String path) {
        DLinkZkUtils zkUtils = DLinkZkUtils.get();

        byte[] data = zkUtils.zkClient().readData(path, true);

        return data;
    }

}
